package com.smart.thrift.model.server;

import com.smart.thrift.config.server.TRegisterProcessor;
import com.smart.thrift.config.server.TRegisterProcessorFactory;
import com.smart.thrift.discoveryandbalance.discovery.Registry;
import com.smart.thrift.discoveryandbalance.discovery.ThriftServerNode;
import com.smart.thrift.exception.server.ThriftServerException;
import com.smart.thrift.model.common.ThriftServerProperties;
import com.smart.thrift.model.wrapper.ThriftServiceWrapper;
import lombok.Getter;
import lombok.Setter;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TTransportException;
import org.apache.thrift.transport.layered.TFastFramedTransport;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Getter
@Setter
public class ThriftHsHaServer extends THsHaServer implements IThriftServer {

    private String serverId;
    private String serverName;
    private Registry<?> registry;
    private ThriftServerNode thriftServerNode;
    private List<ThriftServiceWrapper> serviceWrappers;

    public ThriftHsHaServer(Args args) {
        super(args);
    }

    @Override
    public void startUp(){
        this.serve();
    }

    @Override
    public void shutDown() {
        this.stop();
    }

    public static class Args extends THsHaServer.Args {

        private final Map<String, ThriftServiceWrapper> processorMap = new HashMap<>();

        public Args(List<ThriftServiceWrapper> serviceWrappers, ThriftServerProperties properties) throws TTransportException {
            super(new TNonblockingServerSocket(properties.getPort()));

            transportFactory(new TFastFramedTransport.Factory());
            protocolFactory(new TCompactProtocol.Factory());

            ThriftServerProperties.THsHaServerProperties hsHaProperties = properties.getHsHa();
            minWorkerThreads(hsHaProperties.getMinWorkerThreads());
            maxWorkerThreads(hsHaProperties.getMaxWorkerThreads());

            executorService(createInvokerPool(properties));

            try {
                TRegisterProcessor registerProcessor = TRegisterProcessorFactory.registerProcessor(serviceWrappers);

                processorMap.clear();
                processorMap.putAll(registerProcessor.getProcessorMap());

                processor(registerProcessor);
            } catch (Exception e) {
                throw new ThriftServerException("Can not create multiplexed processor for " + serviceWrappers, e);
            }
        }

        private ExecutorService createInvokerPool(ThriftServerProperties properties) {
            ThriftServerProperties.THsHaServerProperties hsHaProperties = properties.getHsHa();

            return new ThreadPoolExecutor(
                    hsHaProperties.getMinWorkerThreads(),
                    hsHaProperties.getMaxWorkerThreads(),
                    hsHaProperties.getKeepAlivedTime(), TimeUnit.MINUTES,
                    new LinkedBlockingQueue<>(properties.getWorkerQueueCapacity()));
        }

        public Map<String, ThriftServiceWrapper> getProcessorMap() {
            return processorMap;
        }
    }

}
